package com.chitang.common.listener;

import cn.hutool.json.JSONUtil;
import com.chitang.common.IMysqlSyncPlugin;
import com.chitang.common.dto.CanalMessage;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Map;

/**
 * Canal MySQL数据同步消费者
 * 监听mq中的消息
 * @author yicol
 */
@Slf4j
public class MysqlMqSyncListener {

    /**
     * MySQL 数据同步 RabbitMQ 交换器
     */
    private final static String EXCHANGE = "canal_exchange";

    /**
     * 任务回调 RabbitMQ 队列
     * <p>前缀 [mysql.database.] + [数据库名称]</p>
     */
//    private final static String QUEUE = "mysql.database.${xsyw.jdbc.database}";
    private final static String QUEUE = "canal_queue";

    /**
     * 任务回调 RabbitMQ 路由键, 需在yml文件配置好
     */
//    private final static String ROUTING_KEY = "${xsyw.jdbc.database}";
    private final static String ROUTING_KEY = "canal_key";

    @Resource(name = "mysqlSyncPluginMap")
    private Map<String, IMysqlSyncPlugin> pluginMap;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = QUEUE),
            key = ROUTING_KEY,
            exchange = @Exchange(value = EXCHANGE, type = ExchangeTypes.TOPIC)))
    @RabbitHandler
    public void handleMessage(Channel channel, Message message) throws IOException {
        String body = new String(message.getBody());
        String queue = message.getMessageProperties().getConsumerQueue();
        log.info("--收到来自 {} 的数据={}", queue, body);
        try {
            CanalMessage canalMessage = JSONUtil.toBean(body, CanalMessage.class);
            // 由于商品表和商家表共用一个数据库，这里会出现一个问题，由于这里接收消息采用轮训的方式，当改动商品表的数据时，
            // 那么可能在商品模块中接收不到该条消息，而消息是被商家模块那边给消费了，在第二次改动商品的时候才会被商品模块那边消费。
            // 这里就比对一下当前收到的消息是否是对应模块需要的消息数据。
            if (!pluginMap.containsKey(canalMessage.getTable())) {
                // 如果不是，就手动再返回队列
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
                return;
            }
            log.info("--消费来自 {} 的数据={}", queue, body);
            if (canalMessage == null) {
                log.error("数据同步消息为空, 同步数据失败");
                return;
            }
            log.info("************* 同步数据开始 **************");
            long start = System.currentTimeMillis();

            log.info("被注册的表名有：{}", JSONUtil.toJsonStr(pluginMap.keySet()));
            log.info("需要的表名是：{}", canalMessage.getTable());

            IMysqlSyncPlugin plugin =
                    pluginMap.get(canalMessage.getTable());
            plugin.process(canalMessage);

            long end = System.currentTimeMillis();
            log.info("************* 同步数据结束 耗时 {} ms **************", end - start);

        } catch (Exception e) {
            log.error("处理MQ消息异常 {} {}", queue, body, e);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        } finally {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }

    }


    public MysqlMqSyncListener() {
        log.info("************* mysql data sync listener has been starting *************");
    }

}
